package xyz.xuetu.www.rpc.server.tcp;

import cn.hutool.core.util.IdUtil;
import io.vertx.core.Vertx;
import io.vertx.core.buffer.Buffer;
import io.vertx.core.net.NetClient;
import io.vertx.core.net.NetSocket;
import xyz.xuetu.www.rpc.RpcApplication;
import xyz.xuetu.www.rpc.model.RpcRequest;
import xyz.xuetu.www.rpc.model.RpcResponse;
import xyz.xuetu.www.rpc.model.ServiceMetaInfo;
import xyz.xuetu.www.rpc.protocol.*;

import java.io.IOException;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;

/**
 * Vert.x TCP 请求客户端
 */
public class VertxTcpClient {

    /**
     * 发送请求
     * @param rpcRequest
     * @param serviceMetaInfo
     * @return
     */
    public static RpcResponse doRequest(RpcRequest rpcRequest, ServiceMetaInfo serviceMetaInfo) throws ExecutionException, InterruptedException {
        Vertx vertx = Vertx.vertx();
        NetClient netClient = vertx.createNetClient();
        CompletableFuture<RpcResponse> responseFuture = new CompletableFuture<>();
        netClient.connect(serviceMetaInfo.getServicePort(), serviceMetaInfo.getServiceHost(), result -> {
            if(!result.succeeded()) {
                System.err.println("Failed to connect to TCP server");
                return;
            }
            System.out.println("Connected to TCP server");
            NetSocket netSocket = result.result();
            // 构造消息体
            ProtocolMessage<RpcRequest> requestProtocolMessage = new ProtocolMessage<>();
            ProtocolMessage.Header header = new ProtocolMessage.Header();
            header.setMagic(ProtocolConstant.PROTOCOL_MAGIC);
            header.setVersion(ProtocolConstant.PROTOCOL_VERSION);
            ProtocolMessageSerializerEnum serializerEnum = ProtocolMessageSerializerEnum.getEnumByValue(RpcApplication.getRpcConfig().getSerializer());
            if(serializerEnum == null) {
                throw new RuntimeException("序列化协议不存在");
            }
            header.setSerializer((byte) serializerEnum.getKey());
            header.setType((byte) ProtocolMessageTypeEnum.REQUEST.getKey());
            // 生成全局请求 ID
            header.setRequestId(IdUtil.getSnowflakeNextId());
            requestProtocolMessage.setHeader(header);
            requestProtocolMessage.setBody(rpcRequest);

            // 编码请求
            try {
                Buffer encodeBuffer = ProtocolMessageEncoder.encode(requestProtocolMessage);
                netSocket.write(encodeBuffer);
            } catch (IOException e) {
                throw new RuntimeException("协议消息编码错误", e);
            }

            // 接收响应
            netSocket.handler(new TcpBufferHandlerWrapper(buffer -> {
                try {
                    ProtocolMessage<RpcResponse> responseProtocolMessage = (ProtocolMessage<RpcResponse>) ProtocolMessageDecoder.decoder(buffer);
                    responseFuture.complete(responseProtocolMessage.getBody());
                } catch (IOException e) {
                    throw new RuntimeException("协议消息解码错误", e);
                }
            }));
        });
        RpcResponse rpcResponse = responseFuture.get();
        // 关闭连接
        netClient.close();
        return rpcResponse;
    }
}
